package org.poem.service.impl;

import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.google.common.collect.Maps;
import org.poem.ExecTaskDetailPlanVO;
import org.poem.IdsVo;
import org.poem.TaskDetailAvailabilityMaps;
import org.poem.config.KafkaProducer;
import org.poem.datasource.DataAndParametersVo;
import org.poem.datasource.DataSourceVo;
import org.poem.entity.TDatasource;
import org.poem.entity.TTaskDetail;
import org.poem.entity.TTaskDetailTable;
import org.poem.enums.EnumDataType;
import org.poem.enums.KafkaEnums;
import org.poem.id.SnowFlake;
import org.poem.mapper.TTaskDetailMapper;
import org.poem.result.PageableResult;
import org.poem.service.TTaskDetailService;
import org.poem.service.TTaskDetailTableService;
import org.poem.task.TTaskDetailVO;
import org.poem.tool.GatherBuilder;
import org.poem.tool.GatherConnection;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * @author poem
 */
@Service
public class TTaskDetailServiceImpl extends ServiceImpl<TTaskDetailMapper, TTaskDetail> implements TTaskDetailService {

    private static final Logger logger = LoggerFactory.getLogger(TTaskDetailServiceImpl.class);

    @Autowired
    private MysqlGatherData mysqlGatherData;

    @Autowired
    private TTaskDetailTableService tTaskDetailTableService;

    @Autowired
    private TDataSourceServiceImpl dataSourceService;

    @Autowired
    private KafkaProducer kafkaProducer;

    /**
     * @return
     */
    @Override
    public List<TTaskDetail> getAllTaskDetail() {
        QueryWrapper<TTaskDetail> detailQueryWrapper = new QueryWrapper<>();
        detailQueryWrapper.lambda().eq(true, TTaskDetail::getAvailability, 1);
        detailQueryWrapper.lambda().eq(true, TTaskDetail::getStatus, 1);
        return this.list(detailQueryWrapper);
    }

    /**
     * @return
     */
    @Override
    public List<TTaskDetail> getFullTaskDetail() {
        QueryWrapper<TTaskDetail> detailQueryWrapper = new QueryWrapper<>();
        detailQueryWrapper.lambda().eq(true, TTaskDetail::getStatus, 1);
        detailQueryWrapper.lambda().eq(true, TTaskDetail::getComplete, 0);
        detailQueryWrapper.lambda().eq(true, TTaskDetail::getAvailability, 1);
        return this.list(detailQueryWrapper);
    }

    /**
     * @param taskId   状态id
     * @param complete 1 全量同步过 0 没有全量同步
     */
    @Override
    public void updateTaskComplete(Long taskId, Integer complete) {
        TTaskDetail detail = getById(taskId);
        if (detail != null) {
            detail.setComplete(complete);
            this.updateById(detail);
        }
    }

    /**
     * @param id
     */
    @Override
    public void unavailable(Long id) {
        TTaskDetail detail = getById(id);
        if (detail != null) {
            detail.setAvailability(0);
            List<String> tableByTaskId = this.tTaskDetailTableService.getTableByTaskId(id);
            for (String s : tableByTaskId) {
                TaskDetailAvailabilityMaps.add(s, 0);
            }

            this.updateById(detail);
        }
    }

    /**
     * @param id
     */
    @Override
    public void availability(Long id) {
        TTaskDetail detail = getById(id);
        if (detail != null) {
            detail.setAvailability(1);
            List<String> tableByTaskId = this.tTaskDetailTableService.getTableByTaskId(id);
            for (String s : tableByTaskId) {
                TaskDetailAvailabilityMaps.add(s, 1);
            }

            this.updateById(detail);
        }
    }

    /**
     * 任务管理
     *
     * @param pageSize
     * @param pageNumber
     * @return
     */
    @Override
    public PageableResult<TTaskDetailVO> getTaskDetailPage(Integer pageSize, Integer pageNumber) {
        //查询条件对象
        QueryWrapper<TTaskDetail> ew = new QueryWrapper<>();

        List<TDatasource> list = this.dataSourceService.list();
        Map<Long, TDatasource> datasourceMap = Maps.newConcurrentMap();
        for (TDatasource tDatasource : list) {
            datasourceMap.put(tDatasource.getId(), tDatasource);
        }
        //分页插件
        IPage<TTaskDetail> tDataSourceIPage = this.page(new Page<>(pageNumber, pageSize), ew);
        PageableResult<TTaskDetailVO> pageableResult = new PageableResult<>();
        pageableResult.setCurrent(tDataSourceIPage.getCurrent());
        pageableResult.setSize(tDataSourceIPage.getSize());
        pageableResult.setTotal(tDataSourceIPage.getTotal());
        //将查询出来的值set到VO对象中
        DateTimeFormatter df = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
        pageableResult.setVoList(tDataSourceIPage.getRecords().stream().map(
                o -> {
                    TTaskDetailVO dataSourceVo = new TTaskDetailVO();
                    dataSourceVo.setContent(o.getContent());
                    if (o.getUpdateTime() != null) {
                        dataSourceVo.setUpdateTime(o.getUpdateTime().format(df));
                    }
                    if (o.getCreateTime() != null) {
                        dataSourceVo.setCreateTime(o.getCreateTime().format(df));
                    }
                    dataSourceVo.setStatus(o.getStatus());
                    dataSourceVo.setTargetId(o.getTargetDatasourceId());
                    TDatasource targetDatasource = datasourceMap.get(o.getTargetDatasourceId());
                    if (targetDatasource != null) {
                        dataSourceVo.setTargetName(targetDatasource.getName());
                    }
                    dataSourceVo.setDatasourceId(o.getSourceDatasourceId());
                    TDatasource sourceDatasource = datasourceMap.get(o.getSourceDatasourceId());
                    if (sourceDatasource != null) {
                        dataSourceVo.setDatasourceName(sourceDatasource.getName());
                    }

                    dataSourceVo.setName(o.getName());
                    dataSourceVo.setId(o.getId());
                    dataSourceVo.setContent(o.getContent());
                    dataSourceVo.setComplete(o.getComplete());
                    dataSourceVo.setAvailability(o.getAvailability());
                    return dataSourceVo;
                }
        ).collect(Collectors.toList()));
        return pageableResult;
    }

    /**
     * @param id id
     * @return
     */
    @Override
    public TTaskDetailVO getTTaskDetailById(Long id) {
        TTaskDetail o = getById(id);
        DateTimeFormatter df = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
        TTaskDetailVO dataSourceVo = new TTaskDetailVO();
        dataSourceVo.setContent(o.getContent());
        if (o.getUpdateTime() != null) {
            dataSourceVo.setUpdateTime(o.getUpdateTime().format(df));
        }
        dataSourceVo.setTableNames(this.tTaskDetailTableService.getTableByTaskId(id));
        dataSourceVo.setStatus(o.getStatus());
        dataSourceVo.setName(o.getName());
        dataSourceVo.setId(o.getId());
        dataSourceVo.setContent(o.getContent());
        dataSourceVo.setComplete(o.getComplete());
        dataSourceVo.setAvailability(o.getAvailability());
        return dataSourceVo;
    }

    /**
     * @param tTaskDetailVO
     * @return
     */
    @Override
    public String saveOrUpdateTaskDetail(TTaskDetailVO tTaskDetailVO) {
        TTaskDetail detail;
        if (tTaskDetailVO.getId() == null) {
            detail = new TTaskDetail();
            detail.setId(SnowFlake.genLongId());
            detail.setCreateTime(LocalDateTime.now());
            detail.setStatus(1);
            detail.setComplete(0);
            detail.setFlag(true);
            detail.setComplete(0);
        } else {
            detail = getById(tTaskDetailVO.getId());
        }
        detail.setContent(tTaskDetailVO.getContent());
        detail.setName(tTaskDetailVO.getName());
        detail.setTargetDatasourceId(tTaskDetailVO.getTargetId());
        detail.setSourceDatasourceId(tTaskDetailVO.getDatasourceId());
        detail.setUpdateTime(LocalDateTime.now());
        this.tTaskDetailTableService.saveTable(tTaskDetailVO.getTableNames(), detail.getId());
        if (tTaskDetailVO.getId() == null) {
            this.save(detail);

        } else {
            this.updateById(detail);
        }
        return null;
    }

    /**
     * 删除
     *
     * @param idsVo
     * @return
     */
    @Override
    public String deleteTask(IdsVo idsVo) {
        QueryWrapper<TTaskDetail> queryWrapper = new QueryWrapper<>();
        queryWrapper.lambda().in(TTaskDetail::getId, idsVo.getIds());
        this.remove(queryWrapper);
        QueryWrapper<TTaskDetailTable> tTaskDetailTableQueryWrapper = new QueryWrapper<>();
        tTaskDetailTableQueryWrapper.lambda().in(TTaskDetailTable::getTaskDetailId, idsVo.getIds());
        this.tTaskDetailTableService.remove(tTaskDetailTableQueryWrapper);
        return null;
    }

    /**
     * 获取数据
     *
     * @param dataSourceVo
     * @return
     */
    @Override
    public List<String> getSchema(DataAndParametersVo dataSourceVo) {
        logger.info(" init getSchema :" + JSONObject.toJSONString(dataSourceVo));
        DataSourceVo da = dataSourceVo.getDataSourceVo();
        GatherConnection connection = new GatherBuilder()
                .sourceType(dataSourceVo.getDataSourceVo().getSourceType())
                .ip(da.getIp())
                .password(da.getPassword())
                .port(da.getPort())
                .user(da.getUserName())
                .builder();
        return mysqlGatherData.getSchema(connection.getConnection(), dataSourceVo.getDataSourceVo().getSourceType());
    }

    /**
     * 开启数据同步
     * @param taskId
     */
    @Override
    public void startCompleteStartTask(Long taskId) {
        TTaskDetail tTaskDetail = getById(taskId);

        List<String> tableByTaskId = this.tTaskDetailTableService.getTableByTaskId(taskId);

        TDatasource tDatasource = this.dataSourceService.getById(tTaskDetail.getSourceDatasourceId());

        for (String s : tableByTaskId) {
            ExecTaskDetailPlanVO execTaskDetailPlanVO = new ExecTaskDetailPlanVO();
            execTaskDetailPlanVO.setWhere("");
            execTaskDetailPlanVO.setTaskDetailId(tTaskDetail.getId());
            execTaskDetailPlanVO.setId(tTaskDetail.getId());

            execTaskDetailPlanVO.setName(tTaskDetail.getName());
            execTaskDetailPlanVO.setTargetTable(s);


            String topic = "";
            if ("mysql".equalsIgnoreCase(tDatasource.getSourceType())) {
                execTaskDetailPlanVO.setEnumDataType(EnumDataType.MYSQL);
                topic = KafkaEnums.DATA_FULL_TRANSFORM_TO_PRODUCER_TOPIC_MYSQL;
            } else if ("sqlserver".equalsIgnoreCase(tDatasource.getSourceType())) {
                execTaskDetailPlanVO.setEnumDataType(EnumDataType.SQLSERVER);
                topic = KafkaEnums.DATA_FULL_TRANSFORM_TO_PRODUCER_TOPIC_SQLSERVER;
            } else if ("postgres".equalsIgnoreCase(tDatasource.getSourceType())) {
                execTaskDetailPlanVO.setEnumDataType(EnumDataType.POSTGRES);
                topic = KafkaEnums.DATA_FULL_TRANSFORM_TO_PRODUCER_TOPIC_POSTGRES;
            }
            execTaskDetailPlanVO.setSourceIp(tTaskDetail.getSourceDatasourceId() + "");
            execTaskDetailPlanVO.setSourcePasswd(tDatasource.getPassword());
            execTaskDetailPlanVO.setSourceSchema(tDatasource.getSourceSchema());
            execTaskDetailPlanVO.setSourceTableName(s);
            execTaskDetailPlanVO.setSourceIp(tDatasource.getIp());
            execTaskDetailPlanVO.setSourceUserName(tDatasource.getUserName());
            execTaskDetailPlanVO.setSourcePort(tDatasource.getPort());
            kafkaProducer.sendMessage(topic, JSONObject.toJSONString(execTaskDetailPlanVO));

        }

    }

}
